package com.atbeijing.D04;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import java.time.Duration;

/**
 * 5s输入几条数据
 */
public class Example2 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                //字符串,时间戳(单位s)
                .socketTextStream("localhost",9999)
                .map(new MapFunction<String, Word>() {
                    @Override
                    public Word map(String value) throws Exception {
                        String[] s = value.split(" ");
                        return new Word(s[0],Long.parseLong(s[1])*1000);
                    }
                })
                //插入水位线
                .assignTimestampsAndWatermarks(
                        //水位线是一种特殊的事件,逻辑时钟
                        //针对乱序流插入水位线
                        //最大延迟时间设置为3s
                        //默认200ms的机器时间插入一次水位线,即逻辑时钟
                        //水位线=观察到的元素中最大的时间戳-最大延迟时间-1ms
                        WatermarkStrategy.<Word>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(new SerializableTimestampAssigner<Word>(){
                            // 抽取时间戳的逻辑,时间戳的单位必须是毫秒
                            // 如果不指定元素中时间戳的字段，程序就不知道事件时间是哪个字段
                            @Override
                            public long extractTimestamp(Word element, long recordTimestamp) {
                                return element.timestamp;
                            }
                        })
                )
                .keyBy(r -> r.word)
                //滚动事件窗口,
                // 第一个窗口窗口只收集事件中时间戳为[0~4999ms]的数据,当水位线=5000ms时窗口关闭
                // 第二个窗口窗口只收集事件中时间戳为[5000ms~9999ms]的数据,当水位线=10000ms时窗口关闭
                //迟到的数据丢弃,早来的数据放到后面的窗口
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                //全量聚合
                .process(new ProcessWindowFunction<Word, String, String, TimeWindow>() {
                    @Override
                    public void process(String s, Context context, Iterable<Word> elements, Collector<String> out) throws Exception {
                        out.collect(context.window().getStart() + "~" + context.window().getEnd() + "窗口中有 " + elements.spliterator().getExactSizeIfKnown() + " 条数据");
                    }
                })
                .print();


        env.execute();
    }

    public static class Word {
        public String word;
        public Long timestamp; // 毫秒时间戳
        public Word() {
        }

        public Word(String word, Long timestamp) {
            this.word = word;
            this.timestamp = timestamp;
        }

        @Override
        public String toString() {
            return "Word{" +
                    "word='" + word + '\'' +
                    ", timestamp=" + timestamp +
                    '}';
        }
    }
}
